package com.surfilter.massdata.spark.task.daystati;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN061;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.URLUtil;

import scala.Tuple2;

public class SAN061Test implements ISparkTask{
private static final Log log = LogFactory.getLog(DomainAnalyzTask_New.class);
	
	private static final long serialVersionUID = 1L;
	private static Pattern IP_PATTERN = Pattern.compile("(\\d{1,3}\\.){3}(\\d{1,3})");
	private String outName;
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames, Map<String, String> commandMap) {

		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		 String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getPreviousDayHour(new Date()),dayStr);
	        DataFrame dataSource=dataFrames.get("day_domain");
	        JavaPairRDD<String, Long> domainCounts=getDomainCounts(dataSource);
	        JavaRDD<SAN061> rdd = createRdd(domainCounts, date);
	        DataFrame df = dataSource.sqlContext().createDataFrame(rdd, SAN061.class);
	    	
	    	outList.add(new OutQueueEntity(outName, df));
		return null;
	}
	private JavaRDD<SAN061> createRdd(JavaPairRDD<String, Long> domainCounts, final Date date) {
		JavaRDD<SAN061> rdd = domainCounts.mapPartitions(new FlatMapFunction<Iterator<Tuple2<String,Long>>, SAN061>() {

			@Override
			public Iterable<SAN061> call(Iterator<Tuple2<String, Long>> it){
				List<SAN061> list = new ArrayList<SAN061>();
				while(it.hasNext()){
					Tuple2<String, Long> tuple = it.next();
					String domain = tuple._1;
					if(StringUtils.isNotBlank(domain)){
		        		SAN061 sa = new SAN061();
		        		sa.setYear(DateUtil.getCurrentYear(date));
		        		sa.setHalf_year(DateUtil.getHalfYear(date));
		        		sa.setQuarter(DateUtil.getQuarter(date));
		        		sa.setMonth(DateUtil.getCurrentMonth(date));
		        		sa.setWeek(DateUtil.getCurrentWeek(date));
		        		sa.setDay(DateUtil.getCurrentDay(date));
		        		sa.setBuss_type("WEBSITE_COUNT_SUFFIX");
		        		sa.setBuss_value(domain);
		        		sa.setCount_value(tuple._2);
		        		sa.setSta_range(0);
		        		list.add(sa);
		        	}
				}
				return list;
			}
		});
		return rdd;
	}
	private JavaPairRDD<String, Long> getDomainCounts(DataFrame dataSource) {
		 JavaRDD<String> domainSuffix=dataSource.toJavaRDD().map(new Function<Row,String>(){
				private static final long serialVersionUID = 1L;
				@Override
				public String call(Row row){
					String result = "";
					
					String topDomain ="";
					try{
						topDomain = row.getAs("top_domain");
						if(StringUtils.isNotBlank(topDomain)){
							if(IP_PATTERN.matcher(topDomain).matches()){//判断是否为IP
								result =  topDomain;
							}else{
								result = topDomain.substring(topDomain.indexOf(".")+1,topDomain.length());
							}
						}else
						{
							if(topDomain.indexOf(".")>0)
							{
								result = topDomain.substring(topDomain.indexOf(".") + 1,topDomain.length());
							}else{
								result = topDomain;
							}
							}
					
						
					}catch(Exception e){
						if(log.isErrorEnabled()){
							log.error(e.getMessage(),e);
						}
					}
					return result + "@" + row.getAs("visit_count");
				}});
	       JavaPairRDD<String,Long> domainPair = domainSuffix.mapToPair(new PairFunction<String,String,Long>(){
				private static final long serialVersionUID = 1L;
				@Override
				public Tuple2<String, Long> call(String value){
					try{
						String[] valueStr = value.split("@");
						String domain = valueStr[0];
						long count = Long.parseLong(valueStr[1]);
						Tuple2<String,Long> tuple = new Tuple2<String,Long>(domain,count);
						return tuple;
					}catch(Exception e){
						if(log.isErrorEnabled()){
							log.error(e.getMessage(),e);
						}
					}
					return new Tuple2<String,Long>("",0L);
				}});
	        JavaPairRDD<String,Long> domainCounts = domainPair.reduceByKey(new Function2<Long,Long,Long>(){
				private static final long serialVersionUID = 1L;
				@Override
				public Long call(Long x, Long y)throws Exception {
					return x+y;
				}});
	return domainCounts;
	}

}
